
// Copyright (c) 2020-present, INSPUR Co, Ltd. All rights reserved.
// This source code is licensed under Apache 2.0 License.

#include "userkey_index.h"

namespace rocksdb {

UserKeyIndex::UserKeyIndex(UserKeyIndexType type) : IUserKeyIndex(type) {
  treeVoidRef_dlink_ = new ConcurrentDLink<TreeVoidRef *>();
}
UserKeyIndex::~UserKeyIndex() { delete treeVoidRef_dlink_; }

TreeVoidRef *UserKeyIndex::getTreeVoidRef(Slice &userKey, VersionNode *node) {
  auto curRef = new TreeVoidRef();
  curRef->setContentList(node);
  void *retRef = nullptr;
  // nextTID is not absolutely greater than key, so add compare logic.
  // if ARTRef not exists, we will put it onto ART tree.
  bool insertSuc = tree_->insertNoReplace(userKey, curRef, retRef);

  // key has ARTRef object on ART tree.
  if (!insertSuc) {
    delete curRef;
    return (TreeVoidRef *)retRef;
  }

  TreeVoidRef *nextRef = nullptr;
  if (retRef != 0) {
    nextRef = (TreeVoidRef *)(retRef);
    if ((nextRef->Next() == nullptr || nextRef->Prev() == nullptr)) {
      if (parseKeyFromTreeVoidRef(nextRef).compare(userKey) <= 0) {
        // next < key  and  next has not inserted into double link list.
        // need redo get larger ARTRef object from ART tree.
        retRef = this->tree_->getGT(userKey);
        if (retRef != 0)
          nextRef = (TreeVoidRef *)(retRef);
        else {
          nextRef = nullptr;
        }
      }
    }
  }
  if (nextRef == nullptr)
    nextRef = treeVoidRef_dlink_->getTail();
  // insert current ARTRef object into double link list.
  treeVoidRef_dlink_->insert(
      curRef, nextRef, [&userKey](TreeVoidRef *a) -> int {
        return parseKeyFromTreeVoidRef(a).compare(userKey);
      });

  return curRef;
}

TreeVoidRef *UserKeyIndex::getGETreeVoidRef(Slice &userKey) {
  void *eorg = tree_->getGE(userKey);
  TreeVoidRef *retRef = nullptr;
  if (eorg != 0) {
    retRef = (TreeVoidRef *)(eorg);
  }

  return retRef;
}

} // namespace rocksdb